package lucene.server.web.task;

import lombok.extern.slf4j.Slf4j;
import lucene.server.cdc.CDCTask;
import lucene.server.common.exception.SearchException;
import lucene.server.common.task.ITask;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;

import java.util.concurrent.ConcurrentHashMap;

@Slf4j
public class TaskPool  {

    private static int core = Runtime.getRuntime().availableProcessors();
    private static ThreadPoolTaskScheduler threadPoolTaskScheduler =  createThreadPoolTaskScheduler();
    // 用来存入线程执行情况, 方便于停止定时任务时使用
    public static ConcurrentHashMap<String, ITask> taskCache= new ConcurrentHashMap<String, ITask>();
    //排队队列
    public static ThreadPoolTaskScheduler createThreadPoolTaskScheduler(){
        ThreadPoolTaskScheduler threadPoolTaskScheduler = new ThreadPoolTaskScheduler();
        // 线程池大小
        threadPoolTaskScheduler.setPoolSize(core*2);
        // 线程名称
        threadPoolTaskScheduler.setThreadNamePrefix("SearchServerExecutor-");
        // 等待时长
        threadPoolTaskScheduler.setAwaitTerminationSeconds(60);
        // 调度器shutdown被调用时等待当前被调度的任务完成
        threadPoolTaskScheduler.setWaitForTasksToCompleteOnShutdown(true);
        threadPoolTaskScheduler.initialize();
        return threadPoolTaskScheduler;
    }
    public static void runTask(String taskName) {

       if(taskCache.containsKey(taskName)){
            ITask taskTemp =  taskCache.get(taskName);
            if (taskTemp.isRunning()){
                throw  new SearchException("任务正在执行中,请勿重复执行。");
            }
            if (taskTemp instanceof SQLTask){
                threadPoolTaskScheduler.execute((Runnable) taskTemp);
            }else{
                throw  new SearchException("只有SQL任务才能手动运行。");
            }
       }else{
           throw  new SearchException("SQL任务不存在。");
       }
    }

    public static void addTask(ITask task){
        task.createScheduled(threadPoolTaskScheduler);
        taskCache.put(task.getName(), task);
    }

    public static void shutdown(String taskName) {
        if (!taskCache.containsKey(taskName)) {
            return;
        }
        ITask task = taskCache.get(taskName);
        task.shutdown();
        taskCache.remove(taskName);
        task = null;
    }

    public static void shutdown(){
        log.info("Application Close Task Pool...");
        //1、先停止canal任务
        taskCache.values().forEach(task -> {
            if(task instanceof CDCTask){
               task.shutdown();
            }
        });
        //2、再处理CDC任务里面的缓存数据，保证数据的一致性
        taskCache.values().forEach(task -> {
            if(task instanceof DataChangeTask){
               task.shutdown();
            }
        });
        //2、再处理CDC任务里面的缓存数据，保证数据的一致性
        taskCache.values().forEach(task -> {
            if(task instanceof SQLTask){
                task.shutdown();
            }
        });
        taskCache.clear();
    }
}